/****************************************************
 * 创建人:   robin
 * 创建时间: 2023/9/4.004 13:49
 * 项目名称: cpy
 * 文件名称: CollectCommCallable.java
 * 文件描述: 
 *
 * All rights Reserved, Designed By 德品软件
 * @Copyright:2023-2023
 *
 ********************************************************/
package com.dp.threadmanager;

import com.dp.constant.CollectConstant;
import com.dp.dto.MsgUnitDto;
import com.dp.enumeration.ReturnCodeEnum;
import com.dp.enumeration.ServerPortEnum;
import com.dp.service.HandleMessageService;
import com.dp.utils.BufferUtil;
import com.dp.utils.OtherUtil;
import com.dp.vo.response.ResponseData;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.context.annotation.Configuration;

import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.Callable;

/**
 * 包名称：com.dp.threadmanager
 * 类名称：CollectCommCallable
 * 类描述：
 * 创建人： robin
 * 创建时间：2023/9/4.004 13:49
 */
@Slf4j
@Configuration
public class CollectCommCallable implements Callable<Void> {
    
    private HandleMessageService handleMessageService;
    private ServerPortEnum serverPortEnum = null;
    private Selector selector = null;
    private SocketChannel socketChannel = null;
    private ByteBuffer recvByteBuffer = null;
    private boolean socketStatus = true;
    private String ip = null;
    private int port = 0;
    
    public CollectCommCallable(HandleMessageService handleMessageService) {
        this.handleMessageService = handleMessageService;
    }
    
    public void setSocketChannel(SocketChannel socketChannel) {
        this.socketChannel = socketChannel;
    }
    
    public void setServerPortEnum(ServerPortEnum serverPortEnum) {
        this.serverPortEnum = serverPortEnum;
    }
    
    @Override
    public Void call() throws Exception {
        log.info("CollectCommCallable start[{}]", serverPortEnum.getCode());
    
        try {
            while (null == socketChannel || null == serverPortEnum) {
                Thread.sleep(CollectConstant.SECOND_1); // 休眠1秒
            }
        
            if (0 == start()) {
                process();
            }
        
        } catch (Exception e) {
            e.printStackTrace();
            log.error("CollectCommCallable[{}] failed:{}", serverPortEnum.getCode(), e.getMessage());
        } finally {
            stop();
        }
    
        log.info("CollectCommCallable end[{}]", serverPortEnum.getCode());
        return null;
    }
    
    private int start(){
        int returnCode = 0;
        try {
            ip = ((InetSocketAddress)socketChannel.getRemoteAddress()).getAddress().getHostAddress();
            port = ((InetSocketAddress) socketChannel.getRemoteAddress()).getPort();
            socketStatus = true;
            recvByteBuffer = ByteBuffer.allocate(CollectConstant.MSG_BUF_LEN).order(ByteOrder.BIG_ENDIAN);
            selector = Selector.open();
            socketChannel.configureBlocking(false);   // 设置非阻塞模式
            socketChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE, socketChannel);  // 注册 read|write 事件
    
            handleMessageService.handleNewConnection(serverPortEnum, ip, port);
        } catch (Exception e) {
            e.printStackTrace();
            log.error("CollectCommCallable[{}] start failed:{}", serverPortEnum.getCode(), e.getMessage());
            returnCode = -1;
        }
        return returnCode;
    }
    
    private void process() {
        try {
            while (socketStatus) {
                if (selector.select(CollectConstant.SECOND_1) == 0) {
                    log.debug("CollectCommCallable[{}] no events!", serverPortEnum.getCode());
                    continue;
                }
                
                Set<SelectionKey> selectedKeys = selector.selectedKeys();
                Iterator<SelectionKey> keyIterator = selectedKeys.iterator();
                
                while (keyIterator.hasNext()) {
                    SelectionKey key = keyIterator.next();
                    if (key != null) {
                        if (key.isAcceptable()) {
                            log.debug("CollectCommCallable[{}]:a connection was accepted by a ServerSocketChannel!", serverPortEnum.getCode());
                        } else if (key.isConnectable()) {
                            log.debug("CollectCommCallable[{}]:a connection was established with a remote server!", serverPortEnum.getCode());
                        } else if (key.isReadable()) {
                            log.debug("CollectCommCallable[{}]:a channel is ready for reading!", serverPortEnum.getCode());
                            if (recvData() < 0) {
                                socketStatus = false;
                                break;
                            }
                        } else if (key.isWritable()) {
                            log.debug("CollectCommCallable[{}]:a channel is ready for writing!", serverPortEnum.getCode());
                            if (sendData() < 0) {
                                socketStatus = false;
                                break;
                            }
                        } else {
                            log.debug("CollectCommCallable[{}]:unknow SelectionKey!", serverPortEnum.getCode());
                        }
                    }
                    keyIterator.remove();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("CollectCommCallable[{}] process failed:{}", serverPortEnum.getCode(), e.getMessage());
        }
    }
    
    private void stop(){
        try {
            log.info("CollectCommCallable[{}] stop!", serverPortEnum.getCode());
    
            handleMessageService.clearConnection(serverPortEnum, ip, port);
            
            if (socketChannel != null){
                socketChannel.close();
                socketChannel = null;
            }
            if (selector != null) {
                selector.close();
                selector = null;
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("CollectCommCallable[{}] stop failed:{}", serverPortEnum.getCode(), e.getMessage());
        }
    }
    
    private int recvData() {
        int recvLen = -1;
        try {
            recvByteBuffer.clear(); // 把position设为0，把limit设为capacity，一般在把数据写入Buffer前调用。
            recvLen = socketChannel.read(recvByteBuffer);
            log.debug("CollectCommCallable[{}]:recv {} byte.", serverPortEnum.getCode(), recvLen);
            if (recvLen > 0) {
                MsgUnitDto msgUnitDto = new MsgUnitDto();
                msgUnitDto.setServerPortEnum(serverPortEnum);
                recvByteBuffer.flip();       // 写之后要把limit设为当前position，再把position设为0，一般在从Buffer读出数据前调用。
                msgUnitDto.setByteBuffer(BufferUtil.cloneByteBuffer(recvByteBuffer));
                msgUnitDto.setEquipmentIp(ip);
                msgUnitDto.setPort(port);
                log.debug("CollectCommCallable[{}]:recv {}:{} {} byte:{}", serverPortEnum.getCode(), msgUnitDto.getEquipmentIp(), msgUnitDto.getPort(), recvLen, BufferUtil.bytesToHexString(recvByteBuffer.array(), recvLen, true));
                handleMessageService.pushRecvData(serverPortEnum.getCode(), msgUnitDto);
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("CollectCommCallable[{}] recvData failed:{}", serverPortEnum.getCode(), e.getMessage());
            recvLen = -1;
        }
        return recvLen;
    }
    
    private int sendData() {
        int sendLen = -1;
        try {
            ResponseData<ByteBuffer> rsp = handleMessageService.pollSendData(serverPortEnum.getCode(), OtherUtil.getIpPortString(ip, port));
            if (rsp.getCode().equals(ReturnCodeEnum.SUCCESS.getCode())) {
                sendLen = socketChannel.write(rsp.getData());
                log.debug("CollectCommCallable[{}]:send {}:{} {} byte.", serverPortEnum.getCode(), ip, port, sendLen);
                if (sendLen > 0) {
                    log.debug("CollectCommCallable[{}]:send {}:{} {} byte:{}", serverPortEnum.getCode(), ip, port, sendLen, BufferUtil.bytesToHexString(rsp.getData().array(), sendLen, true));
                }
            } else if (rsp.getCode().equals(ReturnCodeEnum.DEVICE_POLL_SEND_MSG_QUEUE_EMPTY.getCode())) {
                sendLen = 0;
                Thread.sleep(CollectConstant.SECOND_1);
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("CollectCommCallable[{}] sendData failed:{}", serverPortEnum.getCode(), e.getMessage());
            sendLen = -1;
        }
        return sendLen;
    }
}
